mod config;
mod event;
mod service;

pub use config::*;
use tracing::{error, info, warn};

use std::{
    collections::HashSet,
    hash::{DefaultHasher, Hash, Hasher},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use serde::{Deserialize, Serialize};

use libp2p::{
    StreamProtocol, Swarm, SwarmBuilder, dcutr,
    gossipsub::{self},
    identify,
    identity::Keypair,
    kad::{self, QueryId, Quorum, store::MemoryStore},
    mdns, noise, ping, relay,
    request_response::{self, OutboundRequestId, cbor},
    swarm::NetworkBehaviour,
    tcp, yamux,
};
use tokio::{
    fs,
    sync::{mpsc, oneshot},
};

use libp2p::kad::{Record, RecordKey};

use crate::{
    app::{
        FileProcessResult, file_chunk_value, make_file_metadata_path_name,
        store::{self, PublicPublishedFile},
    },
    error::P2pNetworkError,
};

/*
    Flow:
    1.File publisher publishes a file to the network (splits to chunks, new record to Kademlia, start providing keys and publishes the file on the network (if set to public))
    2.Client knows/gets a file ID and starts looking for kademlia peers who owns (selecting the nearest one) this file ID (DHT record key from file)
    3.Request metadata from one of the peers
    4.Start donwlonding chunks parallel:
        1.Lookup closest peers who provides a specific chunk
        2.Download a chunk by simply requesting it
        3.Validate chunk from the already requested meatdata
    5.After all chunks downloaded and validated, create the final file based on metadata (use original file name) and validate the root hash
*/

pub const LOG_TARGET: &str = "app::p2p::P2pService";
pub const GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME: &str = "public_files";

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileChunkDownloadRequest {
    pub file_id: u64,
    pub chunk_id: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FileChunkDownloadResponse {
    Success(FileChunkData),
    Error(String),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileChunkData {
    pub file_id: u64,
    pub chunk_id: usize,
    pub data: Vec<u8>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetadataDownloadRequest {
    pub file_id: u64,
}

impl From<&MetadataDownloadRequest> for RecordKey {
    fn from(request: &MetadataDownloadRequest) -> Self {
        RecordKey::new(&request.file_id.to_be_bytes().to_vec())
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MetadataDownloadResponse {
    Success(Vec<u8>),
    Error(String),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProvideFileChunkRequest {
    pub file_id: u64,
    pub chunk_hash: u64,
    pub chunk_id: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProvideFileRequest {
    pub file_id: u64,
    pub metadata: FileProcessResult,
}

#[derive(Debug)]
pub enum P2pCommand {
    RequestMetadata {
        request: MetadataDownloadRequest,
        result: oneshot::Sender<FileProcessResult>,
    },
    RequestFileChunk {
        request: FileChunkDownloadRequest,
        result: oneshot::Sender<Vec<u8>>,
    },
    ProvideFileChunk {
        request: ProvideFileChunkRequest,
        result: oneshot::Sender<bool>,
    },
    ProvideFile {
        request: ProvideFileRequest,
        result: oneshot::Sender<bool>,
    },
    GetPublicFiles {
        result: mpsc::Sender<PublicPublishedFile>,
    },
}

#[derive(Debug)]
pub struct MetadataDownloadRequestData {
    pub providers_query_id: QueryId,
    pub download_metadata_request_id: Option<OutboundRequestId>,
    pub request: MetadataDownloadRequest,
    pub result: Option<oneshot::Sender<FileProcessResult>>,
}

impl MetadataDownloadRequestData {
    pub fn new(
        providers_query_id: QueryId,
        download_metadata_request_id: Option<OutboundRequestId>,
        request: MetadataDownloadRequest,
        result: Option<oneshot::Sender<FileProcessResult>>,
    ) -> Self {
        Self {
            providers_query_id,
            download_metadata_request_id,
            request,
            result,
        }
    }
}

#[derive(Debug)]
pub struct FileChunkDownloadRequestData {
    pub providers_query_id: QueryId,
    pub download_file_chunk_request_id: Option<OutboundRequestId>,
    pub request: FileChunkDownloadRequest,
    pub result: Option<oneshot::Sender<Vec<u8>>>,
}

impl FileChunkDownloadRequestData {
    pub fn new(
        providers_query_id: QueryId,
        download_file_chunk_request_id: Option<OutboundRequestId>,
        request: FileChunkDownloadRequest,
        result: Option<oneshot::Sender<Vec<u8>>>,
    ) -> Self {
        Self {
            providers_query_id,
            download_file_chunk_request_id,
            request,
            result,
        }
    }
}

#[derive(NetworkBehaviour)]
pub struct P2pNetworkBehavior {
    /// 下面是 libp2p 中常见的网络协议的组件，它们的关系为:
    ///
    /// 节点发现层面：mDNS 和 DHT 负责节点的发现。mDNS 在局域网内快速发现节点，
    /// DHT 在广域网中实现分布式节点发现，两者相互补充，扩大节点发现范围。
    ///
    /// 连接维护与身份验证层面：Ping 协议用于维护节点间连接的活跃状态，
    /// Identify 协议在节点建立连接后进行身份信息交换和验证。
    ///
    /// 连接辅助层面：Relay 和 DCUTR 作为连接辅助手段。
    /// 当节点间无法直接连接时，Relay 提供 TCP 数据中转服务，DCUTR 优化跨数据中心的 UDP 传输。
    ///
    /// 消息传播层面：GossipSub 基于前面组件建立的连接，实现消息在节点间的高效传播。
    ///
    /// 解释:
    /// Ping 指向 连接维护，表明其在维护节点连接活跃状态方面的作用。
    ///
    /// Identify 指向 身份验证，体现其在节点身份验证和信息交换的功能。
    ///
    /// mDNS 和 DHT 指向 节点发现，说明它们在节点发现阶段的作用。
    ///
    /// Relay Client 与 Relay Server 关联，共同指向 连接辅助，表示 Relay 为节点连接提供辅助服务。
    ///
    /// DCUTR 也指向 连接辅助，说明其同样在节点连接方面提供优化。
    ///
    /// 连接维护、身份验证、节点发现 和 连接辅助 的结果都用于 节点连接。
    ///
    /// 成功建立 节点连接 后，GossipSub 基于这些连接实现 消息传播。
    ///

    /// Ping
    /// 功能：定期向其他节点发送 Ping 消息，检查节点是否在线，保持连接的活跃状态，测量节点间的延迟。
    /// 应用场景：用于维护节点间连接的健康状态，确保节点能及时发现其他节点的存活情况。
    ping: ping::Behaviour,
    /// Identify
    /// 功能：节点通过该协议交换自身的身份信息，如公钥、支持的协议、监听地址等，帮助其他节点了解自身情况。
    /// 应用场景：在节点建立连接后，进行身份验证和信息交换，提高网络安全性和节点互操作性。
    identify: identify::Behaviour,
    /// mDNS（多播 DNS）
    /// 功能：在局域网内自动发现其他节点，节点通过多播 DNS 消息相互发现，无需中心服务器配置。
    /// 应用场景：适用于局域网环境，快速发现同一局域网内的其他节点。
    mdns: mdns::Behaviour<mdns::tokio::Tokio>,
    /// DHT（分布式哈希表）
    /// 功能：实现分布式节点发现和存储，节点可通过 DHT 查找其他节点的地址信息，类似于分布式数据库。
    /// 应用场景：广域网环境，节点通过引导节点加入 DHT 网络，之后可在网络中查找其他节点。
    kademlia: kad::Behaviour<MemoryStore>,
    /// Relay（中继）
    /// 功能：当节点因 NAT 等原因无法直接建立连接时，通过中继服务器中转数据，实现节点间通信。
    /// Client 与 Server 区分原因：Relay 服务需要明确角色分工。Relay Server 作为中继服务的提供者，
    /// 监听并等待 Relay Client 的连接请求；Relay Client 作为服务使用者，向 Relay Server
    /// 发起连接请求，请求其帮忙中转数据。其他协议通常不需要这种明确的服务提供者和使用者的角色划分，所以不用区分。
    ///
    /// 应用场景：复杂网络环境，如存在严格 NAT 限制的网络，节点之间无法直接通信的情况。
    relay_server: relay::Behaviour,
    relay_client: relay::client::Behaviour,
    /// GossipSub（基于流言传播的订阅 - 发布协议）
    /// 功能：实现消息的高效传播，节点可订阅特定主题，发布者发布的消息会通过 Gossip 机制传播到所有订阅该主题的节点。
    /// 应用场景：消息广播、实时通信等场景，如聊天应用中的消息传递。
    gossipsub: gossipsub::Behaviour,
    /// DCUTR（数据中心间 UDP 传输中继）
    /// 功能：在数据中心间优化 UDP 传输，通过中继机制改善跨数据中心节点间的连接和通信效率。
    /// 应用场景：跨数据中心的 P2P 网络，用于提升不同数据中心节点间的通信质量。
    dcutr: dcutr::Behaviour,

    /// 文件下载
    file_download: cbor::Behaviour<FileChunkDownloadRequest, FileChunkDownloadResponse>,
    /// 元数据下载
    metadata_download: cbor::Behaviour<MetadataDownloadRequest, MetadataDownloadResponse>,
}

pub struct P2pService<FileStore: store::Store> {
    config: P2pServiceConfig,
    file_publish_receiver: mpsc::Receiver<FileProcessResult>,
    file_store: FileStore,
    command_receiver: mpsc::Receiver<P2pCommand>,

    /// 元数据下载请求队列
    metadata_download_requests: Vec<MetadataDownloadRequestData>,
    /// 文件块下载请求队列
    file_chunk_download_requests: Vec<FileChunkDownloadRequestData>,

    public_files: HashSet<PublicPublishedFile>,
}

impl<FileStore: store::Store> P2pService<FileStore> {
    pub fn new(
        config: P2pServiceConfig,
        file_publish_receiver: mpsc::Receiver<FileProcessResult>,
        command_receiver: mpsc::Receiver<P2pCommand>,
        file_store: FileStore,
    ) -> Self {
        Self {
            config,
            file_publish_receiver,
            file_store,
            command_receiver,
            metadata_download_requests: Vec::new(),
            file_chunk_download_requests: Vec::new(),
            public_files: HashSet::new(),
        }
    }

    /// 获取或生成节点的密钥对。
    async fn key_pair(&self) -> Result<Keypair, P2pNetworkError> {
        if let Ok(data) = fs::read(&self.config.keypair_file).await {
            return Ok(Keypair::from_protobuf_encoding(data.as_slice())?);
        };
        let keypair = Keypair::generate_ed25519();
        let encoded = keypair.to_protobuf_encoding()?;
        let dir =
            self.config
                .keypair_file
                .parent()
                .ok_or(P2pNetworkError::FailedToGetKeypairFileDir(
                    self.config.keypair_file.to_path_buf(),
                ))?;
        let _ = fs::remove_file(&self.config.keypair_file).await;
        fs::create_dir_all(dir).await?;
        fs::write(&self.config.keypair_file, encoded).await?;
        Ok(keypair)
    }

    /// 构建 P2P 网络的 Swarm 实例。
    async fn swarm(&self) -> Result<Swarm<P2pNetworkBehavior>, P2pNetworkError> {
        let keypair = self.key_pair().await?;

        // 构建 Swarm 实例，Swarm 是 libp2p 的核心管理结构，负责管理网络连接、事件处理等。

        // Noise 加密配置。
        let noise_config = noise::Config::new;
        // Yamux 多路复用默认配置。
        let yamux_config = yamux::Config::default;
        // TCP 传输默认配置。
        let tcp_config = tcp::Config::default();
        let swarm = SwarmBuilder::with_existing_identity(keypair)
            // 使用 tokio 异步运行时，处理网络事件和异步任务。
            .with_tokio()
            // 配置 TCP 传输层。
            .with_tcp(
                // 使用默认的 TCP 配置。
                tcp_config,
                // 使用 Noise 加密协议，保证通信的安全性。
                noise_config,
                // 使用默认的 Yamux 多路复用配置，在单个连接上复用多个流。
                yamux_config,
            )?
            .with_quic()
            .with_relay_client(noise_config, yamux_config)?
            .with_behaviour(|key_pair, relay_client| {
                // 获取节点的 PeerID。
                let peer_id = key_pair.public().to_peer_id();
                // 创建 Kademlia 协议配置实例，指定协议标识符。
                let mut kad_config = kad::Config::new(StreamProtocol::new("/dfs/1.0.0/kad"));
                // 设置定期引导间隔为 30 秒，定期尝试连接引导节点，更新节点路由表。
                kad_config.set_periodic_bootstrap_interval(Some(Duration::from_secs(30)));

                // 配置 GossipSub 协议的参数。
                let gossipsub_config = gossipsub::ConfigBuilder::default()
                    // 设置心跳间隔为 10 秒，定期维护消息传播网络。
                    .heartbeat_interval(Duration::from_secs(10))
                    // 设置消息验证模式为严格模式，确保消息的合法性。
                    .validation_mode(gossipsub::ValidationMode::Strict)
                    // 自定义消息 ID 生成函数，确保消息 ID 的唯一性。
                    .message_id_fn(|message| {
                        let mut hasher = DefaultHasher::new();
                        // 使用消息数据、主题、来源 PeerID 和当前时间生成哈希值。
                        message.data.hash(&mut hasher);
                        message.topic.hash(&mut hasher);
                        if let Some(peer_id) = message.source {
                            peer_id.hash(&mut hasher);
                        };
                        let now = SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .unwrap()
                            .as_millis();
                        now.to_string().hash(&mut hasher);
                        // 将哈希值转换为消息 ID。
                        gossipsub::MessageId::from(hasher.finish().to_string())
                    })
                    .build()?;

                // 配置 P2pNetworkBehavior 实例。
                Ok(P2pNetworkBehavior {
                    // 初始化 Ping 行为。
                    ping: ping::Behaviour::new(
                        // 使用默认的 Ping 配置。
                        ping::Config::default(),
                    ),
                    // 初始化mDNS行为，用于局域网节点发现
                    mdns: mdns::Behaviour::new(
                        mdns::Config::default(), // 默认mDNS配置
                        peer_id,                 // 使用节点的公钥生成Peer ID
                    )?,
                    identify: identify::Behaviour::new(identify::Config::new(
                        // 初始化身份验证行为。
                        "/dfs/1.0.0".to_string(),
                        // 版本信息。
                        key_pair.public(),
                        // 节点的公钥。
                    )),

                    // 初始化 Kademlia 协议行为。
                    kademlia: kad::Behaviour::with_config(
                        // 节点的 PeerID。
                        peer_id,
                        // 使用内存存储来存储 Kademlia 协议的数据。
                        MemoryStore::new(peer_id),
                        // 使用前面创建的 Kademlia 协议配置。
                        kad_config,
                    ),
                    // 使用前面创建的 Relay 客户端行为实例。
                    relay_client,
                    // 初始化 Relay 服务器行为，使用默认配置。
                    relay_server: relay::Behaviour::new(peer_id, relay::Config::default()),
                    // 初始化 DCUTR 行为。
                    dcutr: dcutr::Behaviour::new(peer_id),
                    // 初始化 GossipSub 行为。
                    gossipsub: gossipsub::Behaviour::new(
                        // 使用签名验证消息的真实性。
                        gossipsub::MessageAuthenticity::Signed(key_pair.clone()),
                        // 使用前面创建的 GossipSub 配置。
                        gossipsub_config,
                    )?,
                    // 文件下载
                    file_download: cbor::Behaviour::new(
                        [(
                            StreamProtocol::new("/dfs/1.0.0/file-download"),
                            request_response::ProtocolSupport::Full,
                        )],
                        request_response::Config::default(),
                    ),

                    // 元数据下载
                    metadata_download: cbor::Behaviour::new(
                        [(
                            StreamProtocol::new("/dfs/1.0.0/metadata-download"),
                            request_response::ProtocolSupport::Full,
                        )],
                        request_response::Config::default(),
                    ),
                })
            })
            .map_err(|err| P2pNetworkError::Libp2pBuilderError(err.to_string()))?
            .with_swarm_config(|config| {
                config.with_idle_connection_timeout(Duration::from_secs(30))
            })
            .build();

        Ok(swarm)
    }

    async fn handle_file_publish(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        file_process_result: FileProcessResult,
    ) {
        match Record::try_from(&file_process_result) {
            Ok(record) => {
                let record_key = record.key.clone();
                if let Err(e) = swarm
                    .behaviour_mut()
                    .kademlia
                    .put_record(record, kad::Quorum::Majority)
                // Majority 大多数节点持有该record
                {
                    error!(target: LOG_TARGET, "failed to put a new record to DHT: {:?}", e)
                }
                if let Err(e) = swarm.behaviour_mut().kademlia.start_providing(record_key) {
                    error!(target: LOG_TARGET, "failed to start providing a new record to DHT: {:?}", e)
                }

                // add new published file to a download file store
                if let Err(e) = self
                    .file_store
                    .add_published_file_record(file_process_result.clone().into())
                {
                    error!(target: LOG_TARGET, "failed to add published file record to file store: {:?}", e)
                }

                // start provide all chunks
                self.start_providing_all_chunks(swarm, &file_process_result);
            }
            Err(e) => {
                error!(target: LOG_TARGET, "failed to get file process result key value: {:?}", e);
            }
        }
    }

    /// handling internal P2P request.
    async fn handle_command(&mut self, swarm: &mut Swarm<P2pNetworkBehavior>, command: P2pCommand) {
        match command {
            P2pCommand::RequestMetadata { request, result } => {
                let key = RecordKey::from(&request);
                let query_id = swarm.behaviour_mut().kademlia.get_providers(key);

                info!(target: LOG_TARGET, "request metadata: {:?}, query_id: {:?}", request, query_id);
                self.metadata_download_requests
                    .push(MetadataDownloadRequestData::new(
                        query_id,
                        None,
                        request,
                        Some(result),
                    ));
            }
            P2pCommand::RequestFileChunk { request, result } => {
                match self.file_store.published_file_exists(request.file_id) {
                    Ok(true) => {
                        info!(target: LOG_TARGET, "File {:?} exists", request.file_id);
                        match self
                            .file_store
                            .published_file_metadata_path(request.file_id)
                        {
                            Ok(path) => {
                                info!(target: LOG_TARGET, "File {:?} metadata path: {:?}", request.file_id, path);
                                match fs::read(path).await {
                                    Ok(metadata) => {
                                        info!(target: LOG_TARGET, "File {:?} metadata: {:?}", request.file_id, metadata);
                                        if let Ok(metadata) = FileProcessResult::try_from(metadata)
                                        {
                                            match metadata.chunk_key_value(request.chunk_id) {
                                                Ok(Some((key, value))) => {
                                                    let record = Record::new(key, value);
                                                    let query_id = swarm
                                                        .behaviour_mut()
                                                        .kademlia
                                                        .get_providers(record.key);

                                                    self.file_chunk_download_requests.push(
                                                        FileChunkDownloadRequestData::new(
                                                            query_id,
                                                            None,
                                                            request,
                                                            Some(result),
                                                        ),
                                                    );
                                                }
                                                Ok(None) => {
                                                    warn!(target: LOG_TARGET, "File chunk {:?} does not exist", request.chunk_id);
                                                }
                                                Err(e) => {
                                                    error!(target: LOG_TARGET, "Failed to get file chunk providers: {:?}", e);
                                                }
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                error!(target: LOG_TARGET, "Failed to check file existence: {:?}", e);
                            }
                        }
                    }
                    Ok(false) => {
                        info!(target: LOG_TARGET, "File {:?} does not exist", request.file_id);
                    }
                    Err(e) => {
                        error!(target: LOG_TARGET, "Failed to check file existence: {:?}", e);
                    }
                }
            }
            P2pCommand::ProvideFileChunk { request, result } => {
                match file_chunk_value(request.chunk_id) {
                    Ok(value) => {
                        let record = Record::new(request.chunk_hash.to_be_bytes().to_vec(), value);
                        let record_key = record.key.clone();
                        if let Err(e) = swarm
                            .behaviour_mut()
                            .kademlia
                            .put_record(record, Quorum::Majority)
                        {
                            error!(target: LOG_TARGET, "failed to start put a file chunk record to DHT: {:?}", e);
                            return;
                        }

                        if let Err(e) = swarm.behaviour_mut().kademlia.start_providing(record_key) {
                            error!(target: LOG_TARGET, "failed to start providing a file chunk record to DHT: {:?}", e);
                            return;
                        }

                        if let Err(e) = result.send(true) {
                            error!(target: LOG_TARGET, "failed to send file chunk provide result: {:?}", e)
                        }
                    }
                    Err(e) => {
                        error!(target: LOG_TARGET, "failed to get chunk key value: {:?}", e);
                    }
                }
            }
            P2pCommand::ProvideFile { request, result } => {
                match Record::try_from(&request.metadata) {
                    Ok(record) => {
                        let record_key = record.key.clone();
                        if let Err(e) = swarm
                            .behaviour_mut()
                            .kademlia
                            .put_record(record, kad::Quorum::Majority)
                        // Majority 大多数节点持有该record
                        {
                            error!(target: LOG_TARGET, "failed to put a new record to DHT: {:?}", e);
                            if let Err(e) = result.send(false) {
                                error!(target: LOG_TARGET, "failed to send file provide result: {:?}", e);
                            }
                            return;
                        }
                        if let Err(e) = swarm.behaviour_mut().kademlia.start_providing(record_key) {
                            error!(target: LOG_TARGET, "failed to start providing a new record to DHT: {:?}", e);
                            if let Err(e) = result.send(false) {
                                error!(target: LOG_TARGET, "failed to send file provide result: {:?}", e);
                            }
                            return;
                        }
                        if let Err(e) = result.send(true) {
                            error!(target: LOG_TARGET, "failed to send file provide result: {:?}", e)
                        }
                    }
                    Err(e) => {
                        error!(target: LOG_TARGET, "failed to convert metadata to record: {:?}", e);
                    }
                }
            }
            P2pCommand::GetPublicFiles { result } => {
                // TODO: 这里只是内存中记录了发布文件信息, 如果节点重启, 就需要从数据库读取
                for public_file in &self.public_files {
                    if let Err(e) = result.send(public_file.clone()).await {
                        error!(target: LOG_TARGET, "failed to send public file: {:?}", e);
                    }
                }
            }
        }
    }

    fn find_metadata_download_request_with_query_id(
        &mut self,
        query_id: QueryId,
    ) -> Option<&MetadataDownloadRequestData> {
        self.metadata_download_requests
            .iter()
            .find(|meta| meta.providers_query_id == query_id)
    }

    fn find_metadata_download_request_with_request_id_mut(
        &mut self,
        request_id: OutboundRequestId,
    ) -> Option<&mut MetadataDownloadRequestData> {
        self.metadata_download_requests
            .iter_mut()
            .find(|meta| meta.download_metadata_request_id == Some(request_id))
    }

    fn set_metadata_download_request_request_id(
        &mut self,
        query_id: QueryId,
        request_id: OutboundRequestId,
    ) {
        self.metadata_download_requests
            .iter_mut()
            .filter(|request| request.providers_query_id == query_id)
            .for_each(|request| {
                if request.download_metadata_request_id.is_none() {
                    request.download_metadata_request_id = Some(request_id);
                }
            });
    }

    fn remove_metadata_download_request(&mut self, request_id: OutboundRequestId) {
        self.metadata_download_requests
            .retain(|request| request.download_metadata_request_id != Some(request_id));
    }

    fn find_file_chunk_download_request_with_query_id(
        &mut self,
        query_id: QueryId,
    ) -> Option<&FileChunkDownloadRequestData> {
        self.file_chunk_download_requests
            .iter()
            .find(|meta| meta.providers_query_id == query_id)
    }

    fn find_file_chunk_download_request_with_request_id_mut(
        &mut self,
        request_id: OutboundRequestId,
    ) -> Option<&mut FileChunkDownloadRequestData> {
        self.file_chunk_download_requests
            .iter_mut()
            .find(|meta| meta.download_file_chunk_request_id == Some(request_id))
    }

    fn remove_file_chunk_download_request(&mut self, request_id: OutboundRequestId) {
        self.file_chunk_download_requests
            .retain(|request| request.download_file_chunk_request_id != Some(request_id));
    }

    fn set_file_chunk_download_request_request_id(
        &mut self,
        query_id: QueryId,
        request_id: OutboundRequestId,
    ) {
        self.file_chunk_download_requests
            .iter_mut()
            .filter(|request| request.providers_query_id == query_id)
            .for_each(|request| {
                if request.download_file_chunk_request_id.is_none() {
                    request.download_file_chunk_request_id = Some(request_id);
                }
            });
    }

    fn start_providing_all_chunks(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        metadata_content: &FileProcessResult,
    ) {
        for chunk_id in 0..metadata_content.number_of_chunks {
            self.start_providing_chunk(swarm, metadata_content, chunk_id);
        }
    }

    fn start_providing_chunk(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        metadata_content: &FileProcessResult,
        chunk_id: usize,
    ) {
        match metadata_content.chunk_key_value(chunk_id) {
            Ok(Some((key, value))) => {
                let record = Record::new(key, value);
                let record_key = record.key.clone();
                if let Err(e) = swarm
                    .behaviour_mut()
                    .kademlia
                    .put_record(record, Quorum::Majority)
                {
                    error!(target: LOG_TARGET, "failed to start put a file chunk record to DHT: {:?}", e);
                    return;
                }

                if let Err(e) = swarm.behaviour_mut().kademlia.start_providing(record_key) {
                    error!(target: LOG_TARGET, "failed to start providing a file chunk record to DHT: {:?}", e)
                }
            }
            Ok(None) => {}
            Err(e) => {
                error!(target: LOG_TARGET, "failed to get chunk key value: {:?}", e);
            }
        }
    }

    fn start_providing_all_files(&mut self, swarm: &mut Swarm<P2pNetworkBehavior>) {
        if let Ok(published_files) = self.file_store.fetch_all_published_files() {
            for published_file in published_files {
                info!(target: LOG_TARGET, "Start providing {} on kademlia DHT...", published_file.original_file_name);
                let metadata_path = make_file_metadata_path_name(&published_file.chunks_directory);
                let metadata_content_result = std::fs::read(metadata_path);
                if let Ok(metadata_content_raw) = metadata_content_result {
                    if let Ok(metadata_content) = metadata_content_raw.try_into() {
                        if let Ok(record) = Record::try_from(&metadata_content) {
                            if let Err(e) = swarm
                                .behaviour_mut()
                                .kademlia
                                .put_record(record, kad::Quorum::Majority)
                            // Majority 大多数节点持有该record
                            {
                                error!(target: LOG_TARGET, "failed to put a new record to DHT: {:?}", e)
                            }

                            // we should start providing all the chunks too!
                            // start provide all chunks
                            self.start_providing_all_chunks(swarm, &metadata_content);
                        }
                    }
                }
            }
        }
    }
}
